package streaming.day02

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.ReceiverInputDStream
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * 想让程序关闭后下次启动能够接着运行，可以使用以下方式
  * 程序启动的时候能够读检查点里面的数据
  */
object StreamingUpdateByKeyV2 {
  val checkpointDir = "./ckpt"


  // Function to create and setup a new StreamingContext
  def functionToCreateContext(): StreamingContext = {
    // master 需要 2 个核, 以防止饥饿情况（starvation scenario）.
    // 1一个核用于运行当前程序，一个核用于接收数据
    val conf = new SparkConf()
      .setAppName("StreamingUpdateStateByKey")
      .setMaster("local[*]")

    // 第二个参数为批次时间间隔，多长时间的数据集为一个批次，这个时间不能拍脑门设置，科学合理的设置,一般会比较短
    // 批次时间要参照spark ui -> total delay 时间来设置，批次时间最好要大于或者等于total delay
    val ssc = new StreamingContext(conf, Seconds(2)) // 2秒钟一个批次

    // 设置checkpoint目录, 缓存历史批次数据
    /**
      * Metadata checkpointing - 将定义 streaming 计算的信息保存到容错存储（如 HDFS）中.这用于从运行 streaming 应用程序的 driver 的节点的故障中恢复（稍后详细讨论）. 元数据包括:
      *     Configuration - 用于创建流应用程序的配置.
      *     DStream operations - 定义 streaming 应用程序的 DStream 操作集.
      *     Incomplete batches - 批量的job 排队但尚未完成.
      *
      * Data checkpointing - 将生成的 RDD 保存到可靠的存储.[把当前批次计算出来的最终的单词次数结果存储到checkpoint目录中]
      *     这在一些将多个批次之间的数据进行组合的 状态 变换中是必需的.
      *     在这种转换中, 生成的 RDD 依赖于先前批次的 RDD, 这导致依赖链的长度随时间而增加.
      *     为了避免恢复时间的这种无限增加（与依赖关系链成比例）, 有状态转换的中间 RDD 会定期 checkpoint 到可靠的存储（例如 HDFS）以切断依赖关系链.
      */
    ssc.checkpoint("./ckpt")


    // 接收受数据
    val stream: ReceiverInputDStream[String] = ssc.socketTextStream("10.172.50.12", 44444)
    // 设置checkpoint的时间间隔，官网建议5-10个滑动窗口
    // stream.checkpoint()

    // 计算wordcount 当前批次的数据
    val wordcounts = stream.flatMap(_.split(" ")).map(word => (word, 1))
      .updateStateByKey(updateFunction _) // 不能把这个地方改成reduceByKey()

    wordcounts.print()

    ssc
  }



  def main(args: Array[String]): Unit = {
    // 从checkpoint目录中尝试恢复streamingcontext对象，如果恢复不了，则调用functionToCreateContext 创建一个新的streamingcontext
    val ssc = StreamingContext.getOrCreate(checkpointDir, functionToCreateContext _)

    ssc.start()
    ssc.awaitTermination()
  }

  /**
    *
    * @param newValues 当前批次某个单词出现的值序列集合（a => Seq[Int](1,1,1)）
    * @param runningCount 上一个批次跑出来的某个单词数量 a 之前的总的次数
    * @return
    */
  def updateFunction(newValues: Seq[Int], runningCount: Option[Int]): Option[Int] = {
    val newCount = newValues.sum + runningCount.getOrElse(0)
    Some(newCount)
  }

}
